package com.atguigu.gmall.realtime.joindemo;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/2/13 10:45
 */
public class ConsumeLeftJoinWithSQL_2 extends BaseSQLApp {
    public static void main(String[] args) {
        new ConsumeLeftJoinWithSQL_2().init(6006, 2, "ConsumeLeftJoinWithSQL_2");
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
       // upsert kafka 消费 upsert-kafka写入的结果
        
        tEnv.executeSql("create table t_left_1(" +
                            " id string, " +
                            " sex string, " +
                            " score int, " +
                            " age int," +
                            " primary key (id) not enforced " +
                            ")" + SQLUtil.getDDLUpsertKafka("t_left_1", "json"));
    
        tEnv.sqlQuery("select sum(score) from t_left_1").execute().print();
        
    }
}
